package com.toutiao.vssue

import java.util.Properties

import kafka.javaapi.producer.Producer
import kafka.producer.{KeyedMessage,ProducerConfig}

class ProducerStore(
        brokers:String,
        topic:String
      ) {
  
  private val prop = new Properties
  
  prop.put("metadata.broker.list",brokers)
  prop.put("serializer.class", "kafka.serializer.StringEncoder")
  prop.put("request.required.acks","1")
  
  private val producer = new Producer[String,String](new ProducerConfig(prop))
  
  def send(tasks:Array[String]) = {
    tasks.par.foreach{task =>
      producer.send(new KeyedMessage[String,String](topic,task))
      }
  }
  
}
